package cn.nju.cloudcomputing.streaming;

import cn.nju.cloudcomputing.db.CRUDTemplate;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;
import scala.Tuple2;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @BelongsProject: SparkProject
 * @BelongsPackage: cn.nju.cloudcomputing.streaming
 * @Author: shadowbynl
 * @CreateTime: 2022-11-16  20:34
 * @Description: TODO
 * @Version: 1.0
 */
public class GithubRealTime {
    public static JavaSparkContext sc = null;

    private static final String SERVER_URL = "172.19.240.61:9092,172.19.240.168:9092,172.19.240.191:9092";

    private static final Integer TAKE_NUM = 25;//批次数量

    private static ExecutorService pool = Executors.newFixedThreadPool(10);

    public static void main(String[] arg) throws IOException,
            InterruptedException {
        System.setProperty("HADOOP_USER_NAME","root");
        SparkConf conf = new SparkConf()
                //配置Spark Streaming程序运以本地方式运行
                .setMaster("local[1]") //多个executor会导致foreach输出顺序有问题
                .setAppName("stream_g");
//        SparkConf conf = new SparkConf()
//                //配置Spark Streaming程序运以本地方式运行
//                .setMaster("spark://spark01:7077")
//                .setAppName("stream_ad");
        //配置Spark Streaming程序间隔10秒消费Kafka生产者生产的Github仓库数据
        sc = new JavaSparkContext(conf);
        JavaStreamingContext jsc =
                new JavaStreamingContext(sc, Durations.seconds(10));
        //jsc.checkpoint("hdfs://172.19.240.24:9000/checkpoint");//192.168.229.200

        //指定Kafka消费者的topic
        final Collection<String> topics = Arrays.asList("test-2");
        Map<String, Object> kafkaParams = new HashMap<>();
        //Kafka服务监听端口
//        kafkaParams.put("bootstrap.servers",
//                "spark01:9092,spark02:9092,spark03:9092");
        kafkaParams.put("bootstrap.servers",
                "172.19.240.61:9092");//spark0
//        kafkaParams.put("bootstrap.servers",
//                "spark01:9092");//spark0
        //指定kafka输出key的数据类型及编码格式（默认为字符串类型编码格式为uft-8）
        kafkaParams.put("key.deserializer", StringDeserializer.class);
        //指定kafka输出value的数据类型及编码格式（默认为字符串类型编码格式为uft-8）
        kafkaParams.put("value.deserializer", StringDeserializer.class);
        //消费者组名称,随便指定一个名称即可
        kafkaParams.put("group.id", "ag");
        //表示如果有offset记录就从offset记录开始消费,
        // 如果没有就从最新的数据开始消费,offset是用来记录消费到哪一条数据了
        kafkaParams.put("auto.offset.reset", "earliest");
        //消费者定期自动提交offset到ZooKeeper
        kafkaParams.put("enable.auto.commit", true);
        SimpleDateFormat format=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

        JavaInputDStream<ConsumerRecord<String, String>> githubDataStream =
                KafkaUtils.createDirectStream(
                        jsc,
                        LocationStrategies.PreferConsistent(),
                        ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams)
                );
        /**
         * 1、从kafka读取基本数据流 githubDataStream
         *   ConsumerRecord<String, String>
         *   value为发送时八个字段拼接的数据  3,sauspiel,webpack-rails,Ruby,0,0,0,79,MIT License
         * 2、基于基本数据流
         * 2/1 该批次fork数的相关统计
         *   1 以 3,sauspiel,webpack-rails 为键  fork为值
         *     JavaPairDStream<String,Integer> forkDataStream
         *   2 transform 转换成RDD 进行排序 取top10 持久化
         *   3 foreach 统计 fork>100 持久化
         *
         * 2/2 该批次star数的相关统计
         *     同fork
         *     starDataStream
         *
         * 2/3 该批次issue数的相关统计
         *
         *
         * 2/4 该批次size的相关统计
         */


        /**
         * fork统计
         */

        //fork基本流
        JavaPairDStream<String, Integer> forkDataStream = githubDataStream.mapToPair(new PairFunction<ConsumerRecord<String, String>, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(ConsumerRecord<String, String> record) throws Exception {
                String[] value = record.value().split(",");
                String bid = value[0];
                String user_name = value[1];
                String repo_name = value[2];
                Integer fork = Integer.valueOf(value[4]);
                String key = String.format("%s,%s,%s", bid, user_name, repo_name);
                return new Tuple2<>(key, fork);
            }
        });

        //fork统计持久化流
        forkDataStream.foreachRDD(new VoidFunction<JavaPairRDD<String, Integer>>() {
            @Override
            public void call(JavaPairRDD<String, Integer> forkSortRDD) throws Exception {
                forkSortRDD.foreach(new VoidFunction<Tuple2<String, Integer>>() {
                    @Override
                    public void call(Tuple2<String, Integer> forkSortTuple2) throws Exception {
                        String[] key = forkSortTuple2._1.split(",");
                        Integer bid = Integer.valueOf(key[0]);
                        String user_name = key[1];
                        String repo_name = key[2];
                        Integer cnt = forkSortTuple2._2;
                        String topic = "fork";
                        if (cnt >= 1000) {
                            String sql = "insert into threshold_res(topic, user_name, repo_name, cnt, create_time, bid) values(?, ?, ?, ?, ?, ?)";
                            CRUDTemplate.executeUpdate(sql, topic, user_name, repo_name, cnt, new Date(), bid);
                        }

                    }
                });
            }
        });

        //fork排序流
        JavaPairDStream<String, Integer>  forkSortStream = forkDataStream.transformToPair(new Function<JavaPairRDD<String, Integer>, JavaPairRDD<String, Integer>>() {
            @Override
            public JavaPairRDD<String, Integer> call(JavaPairRDD<String, Integer> forkDataRDD) throws Exception {
                List<Tuple2<String, Integer>> res = forkDataRDD.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
                    @Override
                    public Tuple2<Integer, String> call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                        return new Tuple2<>(stringIntegerTuple2._2, stringIntegerTuple2._1);
                    }
                }).sortByKey(false).mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
                    @Override
                    public Tuple2<String, Integer> call(Tuple2<Integer, String> integerStringTuple2) throws Exception {
//                        System.out.println(integerStringTuple2._2 + "_" + integerStringTuple2._1);
                        return new Tuple2<>(integerStringTuple2._2, integerStringTuple2._1);
                    }
                }).take(TAKE_NUM);
                System.out.println("===============fork排序OK===================");
                System.out.println(res);
                return sc.parallelizePairs(res);
            }
        });


        //fork排序持久化流
        //存在没按顺序插入的问题
        forkSortStream.foreachRDD(new VoidFunction<JavaPairRDD<String, Integer>>() {
            @Override
            public void call(JavaPairRDD<String, Integer> forkSortRDD) throws Exception {
//                forkSortRDD.foreach(new VoidFunction<Tuple2<String, Integer>>() {
//                    @Override
//                    public void call(Tuple2<String, Integer> forkSortTuple2) throws Exception {
//                        String[] key = forkSortTuple2._1.split(",");
//                        Integer bid = Integer.valueOf(key[0]);
//                        String user_name = key[1];
//                        String repo_name = key[2];
//                        Integer cnt = forkSortTuple2._2;
//                        String topic = "fork";
//                        String sql = "insert into top_res(topic, user_name, repo_name, cnt, create_time, bid) values(?, ?, ?, ?, ?, ?)";
//                        CRUDTemplate.executeUpdate(sql, topic, user_name, repo_name, cnt, new Date(), bid);
//                    }
//                });
                forkSortRDD.foreachPartition(new VoidFunction<Iterator<Tuple2<String, Integer>>>() {
                    @Override
                    public void call(Iterator<Tuple2<String, Integer>> tuple2Iterator) throws Exception {
                        while(tuple2Iterator.hasNext()) {
                            Tuple2<String, Integer> forkSortTuple2 = tuple2Iterator.next();
                            String[] key = forkSortTuple2._1.split(",");
                            Integer bid = Integer.valueOf(key[0]);
                            String user_name = key[1];
                            String repo_name = key[2];
                            Integer cnt = forkSortTuple2._2;
                            String topic = "fork";
                            System.out.println(cnt);
                            String sql = "insert into top_res(topic, user_name, repo_name, cnt, create_time, bid) values(?, ?, ?, ?, ?, ?)";
                            System.out.println("++++======插入数据top====");
                            CRUDTemplate.executeUpdate(sql, topic, user_name, repo_name, cnt, new Date(), bid);
                        }
                    }
                });
            }
        });


        /**
         * star统计
         */

        //star基本流
        JavaPairDStream<String, Integer> starDataStream = githubDataStream.mapToPair(new PairFunction<ConsumerRecord<String, String>, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(ConsumerRecord<String, String> record) throws Exception {
                String[] value = record.value().split(",");
                String bid = value[0];
                String user_name = value[1];
                String repo_name = value[2];
                Integer fork = Integer.valueOf(value[5]);
                String key = String.format("%s,%s,%s", bid, user_name, repo_name);
                return new Tuple2<>(key, fork);
            }
        });

        //star统计持久化流
        starDataStream.foreachRDD(new VoidFunction<JavaPairRDD<String, Integer>>() {
            @Override
            public void call(JavaPairRDD<String, Integer> starSortRDD) throws Exception {
                starSortRDD.foreach(new VoidFunction<Tuple2<String, Integer>>() {
                    @Override
                    public void call(Tuple2<String, Integer> starSortTuple2) throws Exception {
                        String[] key = starSortTuple2._1.split(",");
                        Integer bid = Integer.valueOf(key[0]);
                        String user_name = key[1];
                        String repo_name = key[2];
                        Integer cnt = starSortTuple2._2;
                        String topic = "star";
                        // 数量大则插入 threshold
                        if (cnt >= 1000) {
                            String sql = "insert into threshold_res(topic, user_name, repo_name, cnt, create_time, bid) values(?, ?, ?, ?, ?, ?)";
                            CRUDTemplate.executeUpdate(sql, topic, user_name, repo_name, cnt, new Date(), bid);
                        }

                    }
                });
            }
        });

        //star排序流
        JavaPairDStream<String, Integer>  starSortStream = starDataStream.transformToPair(new Function<JavaPairRDD<String, Integer>, JavaPairRDD<String, Integer>>() {
            @Override
            public JavaPairRDD<String, Integer> call(JavaPairRDD<String, Integer> starDataRDD) throws Exception {
                List<Tuple2<String, Integer>> res = starDataRDD.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
                    @Override
                    public Tuple2<Integer, String> call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                        return new Tuple2<>(stringIntegerTuple2._2, stringIntegerTuple2._1);
                    }
                }).sortByKey(false).mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
                    @Override
                    public Tuple2<String, Integer> call(Tuple2<Integer, String> integerStringTuple2) throws Exception {
//                        System.out.println(integerStringTuple2._2 + "_" + integerStringTuple2._1);
                        return new Tuple2<>(integerStringTuple2._2, integerStringTuple2._1);
                    }
                }).take(TAKE_NUM);
                System.out.println("===============star排序OK===================");
                System.out.println(res);
                return sc.parallelizePairs(res);
            }
        });


        //star排序持久化流
        //存在没按顺序插入的问题
        starSortStream.foreachRDD(new VoidFunction<JavaPairRDD<String, Integer>>() {
            @Override
            public void call(JavaPairRDD<String, Integer> starSortRDD) throws Exception {
                starSortRDD.foreachPartition(new VoidFunction<Iterator<Tuple2<String, Integer>>>() {
                    @Override
                    public void call(Iterator<Tuple2<String, Integer>> tuple2Iterator) throws Exception {
                        while(tuple2Iterator.hasNext()) {
                            Tuple2<String, Integer> starSortTuple2 = tuple2Iterator.next();
                            String[] key = starSortTuple2._1.split(",");
                            Integer bid = Integer.valueOf(key[0]);
                            String user_name = key[1];
                            String repo_name = key[2];
                            Integer cnt = starSortTuple2._2;
                            String topic = "star";
                            System.out.println("=========插入star==========");
                            String sql = "insert into top_res(topic, user_name, repo_name, cnt, create_time, bid) values(?, ?, ?, ?, ?, ?)";
                            CRUDTemplate.executeUpdate(sql, topic, user_name, repo_name, cnt, new Date(), bid);
                        }
                    }
                });
            }
        });


        /**
         * issue统计
         */

        //issue基本流
        JavaPairDStream<String, Integer> issueDataStream = githubDataStream.mapToPair(new PairFunction<ConsumerRecord<String, String>, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(ConsumerRecord<String, String> record) throws Exception {
                String[] value = record.value().split(",");
                String bid = value[0];
                String user_name = value[1];
                String repo_name = value[2];
                Integer fork = Integer.valueOf(value[6]);
                String key = String.format("%s,%s,%s", bid, user_name, repo_name);
                return new Tuple2<>(key, fork);
            }
        });

        //issue统计持久化流
        issueDataStream.foreachRDD(new VoidFunction<JavaPairRDD<String, Integer>>() {
            @Override
            public void call(JavaPairRDD<String, Integer> issueSortRDD) throws Exception {
                issueSortRDD.foreach(new VoidFunction<Tuple2<String, Integer>>() {
                    @Override
                    public void call(Tuple2<String, Integer> issueSortTuple2) throws Exception {
                        String[] key = issueSortTuple2._1.split(",");
                        Integer bid = Integer.valueOf(key[0]);
                        String user_name = key[1];
                        String repo_name = key[2];
                        Integer cnt = issueSortTuple2._2;
                        String topic = "issue";
                        if (cnt >= 100) {
                            String sql = "insert into threshold_res(topic, user_name, repo_name, cnt, create_time, bid) values(?, ?, ?, ?, ?, ?)";
                            CRUDTemplate.executeUpdate(sql, topic, user_name, repo_name, cnt, new Date(), bid);
                        }

                    }
                });
            }
        });

        //issue排序流
        JavaPairDStream<String, Integer>  issueSortStream = issueDataStream.transformToPair(new Function<JavaPairRDD<String, Integer>, JavaPairRDD<String, Integer>>() {
            @Override
            public JavaPairRDD<String, Integer> call(JavaPairRDD<String, Integer> issueDataRDD) throws Exception {
                List<Tuple2<String, Integer>> res = issueDataRDD.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
                    @Override
                    public Tuple2<Integer, String> call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                        return new Tuple2<>(stringIntegerTuple2._2, stringIntegerTuple2._1);
                    }
                }).sortByKey(false).mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
                    @Override
                    public Tuple2<String, Integer> call(Tuple2<Integer, String> integerStringTuple2) throws Exception {
//                        System.out.println(integerStringTuple2._2 + "_" + integerStringTuple2._1);
                        return new Tuple2<>(integerStringTuple2._2, integerStringTuple2._1);
                    }
                }).take(TAKE_NUM);
                System.out.println("===============issue排序OK===================");
                System.out.println(res);

                return sc.parallelizePairs(res);
            }
        });


        //issue排序持久化流
        //存在没按顺序插入的问题
        issueSortStream.foreachRDD(new VoidFunction<JavaPairRDD<String, Integer>>() {
            @Override
            public void call(JavaPairRDD<String, Integer> issueSortRDD) throws Exception {
                issueSortRDD.foreachPartition(new VoidFunction<Iterator<Tuple2<String, Integer>>>() {
                    @Override
                    public void call(Iterator<Tuple2<String, Integer>> tuple2Iterator) throws Exception {
                        while(tuple2Iterator.hasNext()) {
                            Tuple2<String, Integer> issueSortTuple2 = tuple2Iterator.next();
                            String[] key = issueSortTuple2._1.split(",");
                            Integer bid = Integer.valueOf(key[0]);
                            String user_name = key[1];
                            String repo_name = key[2];
                            Integer cnt = issueSortTuple2._2;
                            String topic = "issue";
                            System.out.println(cnt);
                            String sql = "insert into top_res(topic, user_name, repo_name, cnt, create_time, bid) values(?, ?, ?, ?, ?, ?)";
                            CRUDTemplate.executeUpdate(sql, topic, user_name, repo_name, cnt, new Date(), bid);
                        }
                    }
                });
            }
        });


        /**
         * size统计
         */

        //size基本流
        JavaPairDStream<String, Integer> sizeDataStream = githubDataStream.mapToPair(new PairFunction<ConsumerRecord<String, String>, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(ConsumerRecord<String, String> record) throws Exception {
                String[] value = record.value().split(",");
                String bid = value[0];
                String user_name = value[1];
                String repo_name = value[2];
                Integer fork = Integer.valueOf(value[7]);
                String key = String.format("%s,%s,%s", bid, user_name, repo_name);
                return new Tuple2<>(key, fork);
            }
        });

        //size统计持久化流
        sizeDataStream.foreachRDD(new VoidFunction<JavaPairRDD<String, Integer>>() {
            @Override
            public void call(JavaPairRDD<String, Integer> sizeSortRDD) throws Exception {
                sizeSortRDD.foreach(new VoidFunction<Tuple2<String, Integer>>() {
                    @Override
                    public void call(Tuple2<String, Integer> sizeSortTuple2) throws Exception {
                        String[] key = sizeSortTuple2._1.split(",");
                        Integer bid = Integer.valueOf(key[0]);
                        String user_name = key[1];
                        String repo_name = key[2];
                        Integer cnt = sizeSortTuple2._2;
                        String topic = "size";
                        if (cnt >= 100000) {
                            String sql = "insert into threshold_res(topic, user_name, repo_name, cnt, create_time, bid) values(?, ?, ?, ?, ?, ?)";
                            CRUDTemplate.executeUpdate(sql, topic, user_name, repo_name, cnt, new Date(), bid);
                        }

                    }
                });
            }
        });

        //size排序流
        JavaPairDStream<String, Integer>  sizeSortStream = sizeDataStream.transformToPair(new Function<JavaPairRDD<String, Integer>, JavaPairRDD<String, Integer>>() {
            @Override
            public JavaPairRDD<String, Integer> call(JavaPairRDD<String, Integer> sizeDataRDD) throws Exception {
                List<Tuple2<String, Integer>> res = sizeDataRDD.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
                    @Override
                    public Tuple2<Integer, String> call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                        return new Tuple2<>(stringIntegerTuple2._2, stringIntegerTuple2._1);
                    }
                }).sortByKey(false).mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
                    @Override
                    public Tuple2<String, Integer> call(Tuple2<Integer, String> integerStringTuple2) throws Exception {
//                        System.out.println(integerStringTuple2._2 + "_" + integerStringTuple2._1);
                        return new Tuple2<>(integerStringTuple2._2, integerStringTuple2._1);
                    }
                }).take(TAKE_NUM);
                System.out.println("===============size排序OK===================");
                System.out.println(res);
                return sc.parallelizePairs(res);
            }
        });


        //size排序持久化流
        //存在没按顺序插入的问题
        sizeSortStream.foreachRDD(new VoidFunction<JavaPairRDD<String, Integer>>() {
            @Override
            public void call(JavaPairRDD<String, Integer> sizeSortRDD) throws Exception {
                sizeSortRDD.foreachPartition(new VoidFunction<Iterator<Tuple2<String, Integer>>>() {
                    @Override
                    public void call(Iterator<Tuple2<String, Integer>> tuple2Iterator) throws Exception {
                        while(tuple2Iterator.hasNext()) {
                            Tuple2<String, Integer> sizeSortTuple2 = tuple2Iterator.next();
                            String[] key = sizeSortTuple2._1.split(",");
                            Integer bid = Integer.valueOf(key[0]);
                            String user_name = key[1];
                            String repo_name = key[2];
                            Integer cnt = sizeSortTuple2._2;
                            String topic = "size";
                            System.out.println(cnt);
                            String sql = "insert into top_res(topic, user_name, repo_name, cnt, create_time, bid) values(?, ?, ?, ?, ?, ?)";
                            CRUDTemplate.executeUpdate(sql, topic, user_name, repo_name, cnt, new Date(), bid);
                        }
                    }
                });
            }
        });

        /**
         * language统计
         */
        //language基本流
        JavaPairDStream<String, Integer> langDataStream = githubDataStream.mapToPair(new PairFunction<ConsumerRecord<String, String>, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(ConsumerRecord<String, String> record) throws Exception {
                String[] value = record.value().split(",");
                String lang = value[0] + ',' + value[3];
                return new Tuple2<>(lang, 1);
            }
        });

        JavaPairDStream<Tuple2<String, Integer>, Long> langCntDataStream = langDataStream.countByValue();
        JavaPairDStream<String, Integer> langCntDataStream2 = langCntDataStream.mapToPair(new PairFunction<Tuple2<Tuple2<String, Integer>, Long>, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(Tuple2<Tuple2<String, Integer>, Long> tuple2LongTuple2) throws Exception {
                return new Tuple2<>(tuple2LongTuple2._1._1, tuple2LongTuple2._2.intValue());
            }
        });

        //排序 top10
        //lang排序流
        JavaPairDStream<String, Integer>  langSortStream =
                langCntDataStream2.transformToPair(new Function<JavaPairRDD<String, Integer>, JavaPairRDD<String, Integer>>() {
                    @Override
                    public JavaPairRDD<String, Integer> call(JavaPairRDD<String, Integer> langDataRDD) throws Exception {
                        List<Tuple2<String, Integer>> res = langDataRDD.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
                            @Override
                            public Tuple2<Integer, String> call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                                return new Tuple2<>(stringIntegerTuple2._2, stringIntegerTuple2._1);
                            }
                        }).sortByKey(false).mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
                            @Override
                            public Tuple2<String, Integer> call(Tuple2<Integer, String> integerStringTuple2) throws Exception {
                                System.out.println(integerStringTuple2._2 + "_" + integerStringTuple2._1);
                                return new Tuple2<>(integerStringTuple2._2, integerStringTuple2._1);
                            }
                        }).take(11);
                        System.out.println(res);
                        return sc.parallelizePairs(res);
                    }
                });


        //lang排序持久化流
        //存在没按顺序插入的问题
        langSortStream.foreachRDD(new VoidFunction<JavaPairRDD<String, Integer>>() {
            @Override
            public void call(JavaPairRDD<String, Integer> langSortRDD) throws Exception {
                langSortRDD.foreachPartition(new VoidFunction<Iterator<Tuple2<String, Integer>>>() {
                    @Override
                    public void call(Iterator<Tuple2<String, Integer>> tuple2Iterator) throws Exception {
                        while(tuple2Iterator.hasNext()) {
                            Tuple2<String, Integer> langSortTuple2 = tuple2Iterator.next();
                            String[] key = langSortTuple2._1.split(",");
                            Integer bid = Integer.valueOf(key[0]);
                            String lang = key[1];
                            Integer cnt = langSortTuple2._2;
                            if (lang != null && !"null".equals(lang)) {
                                String sql = "insert into lang_res(language, cnt, create_time, bid) values(?, ?, ?, ?)";
                                CRUDTemplate.executeUpdate(sql, lang, cnt, new Date(), bid);
                            }
                        }
                    }
                });
            }
        });


        /**
         * license统计
         */
        //language基本流
        JavaPairDStream<String, Integer> licenseDataStream = githubDataStream.mapToPair(new PairFunction<ConsumerRecord<String, String>, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(ConsumerRecord<String, String> record) throws Exception {
                String[] value = record.value().split(",");
                String lang = value[0] + ',' + value[8];
                return new Tuple2<>(lang, 1);
            }
        });

        JavaPairDStream<Tuple2<String, Integer>, Long> licenseCntDataStream = licenseDataStream.countByValue();
        JavaPairDStream<String, Integer> licenseCntDataStream2 = licenseCntDataStream.mapToPair(new PairFunction<Tuple2<Tuple2<String, Integer>, Long>, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(Tuple2<Tuple2<String, Integer>, Long> tuple2LongTuple2) throws Exception {
                return new Tuple2<>(tuple2LongTuple2._1._1, tuple2LongTuple2._2.intValue());
            }
        });

        //排序 top10
        //license排序流
        JavaPairDStream<String, Integer>  licenseSortStream =
                licenseCntDataStream2.transformToPair(new Function<JavaPairRDD<String, Integer>, JavaPairRDD<String, Integer>>() {
                    @Override
                    public JavaPairRDD<String, Integer> call(JavaPairRDD<String, Integer> licenseDataRDD) throws Exception {
                        List<Tuple2<String, Integer>> res = licenseDataRDD.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
                            @Override
                            public Tuple2<Integer, String> call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                                return new Tuple2<>(stringIntegerTuple2._2, stringIntegerTuple2._1);
                            }
                        }).sortByKey(false).mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
                            @Override
                            public Tuple2<String, Integer> call(Tuple2<Integer, String> integerStringTuple2) throws Exception {
                                System.out.println(integerStringTuple2._2 + "_" + integerStringTuple2._1);
                                return new Tuple2<>(integerStringTuple2._2, integerStringTuple2._1);
                            }
                        }).take(11);
                        System.out.println(res);
                        return sc.parallelizePairs(res);
                    }
                });


        //license排序持久化流
        licenseSortStream.foreachRDD(new VoidFunction<JavaPairRDD<String, Integer>>() {
            @Override
            public void call(JavaPairRDD<String, Integer> licenseSortRDD) throws Exception {
                licenseSortRDD.foreachPartition(new VoidFunction<Iterator<Tuple2<String, Integer>>>() {
                    @Override
                    public void call(Iterator<Tuple2<String, Integer>> tuple2Iterator) throws Exception {
                        while(tuple2Iterator.hasNext()) {
                            Tuple2<String, Integer> langSortTuple2 = tuple2Iterator.next();
                            String[] key = langSortTuple2._1.split(",");
                            Integer bid = Integer.valueOf(key[0]);
                            String license = key[1];
                            Integer cnt = langSortTuple2._2;
                            if (license != null && !"null".equals(license)) {
                                String sql = "insert into license_res(license, cnt, create_time, bid) values(?, ?, ?, ?)";
                                CRUDTemplate.executeUpdate(sql, license, cnt, new Date(), bid);
                            }
                        }
                    }
                });
            }
        });


        jsc.start();
        jsc.awaitTermination();
        jsc.close();
    }

}
